package com.gitee.xhs.iterateStream;

import com.gitee.xhs.dao.IterateTransaction;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import static com.gitee.xhs.OperatorConstant.ITERATOR_OUT_PUT_TAG;

/**
 * @author BigHead
 * @date 2022/7/19 16:48
 * @description IterateStreamMain
 */
public class IterateStreamMain {

    public SingleOutputStreamOperator<IterateTransaction> iterateLogic(SingleOutputStreamOperator<Transaction> mainDataStream) {

        IterativeStream<IterateTransaction> transactionIterativeStream = mainDataStream
                .getSideOutput(ITERATOR_OUT_PUT_TAG)
                .map(
                        new OperatorIterateToIterateTransactionMap()
                )
                .iterate();

        DataStream<IterateTransaction> iterateMainStream = transactionIterativeStream
                .map(
                        new OperatorIterateLoopMap()
                );

        DataStream<IterateTransaction> loopStream = iterateMainStream
                .filter(
                        new OperatorIterateLoopFilter()
                );

        transactionIterativeStream.closeWith(loopStream);

        return iterateMainStream
                .filter(
                        new OperatorIterateEndLoopFilter()
                );
    }

}
